package com.winning.service;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.winning.entity.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.IOException;

/**
 * @Package: com.winning.service
 * @ClassName: Consumer2Service
 * @Author: lujieni
 * @Description: 测试橙联代码
 * @Date: 2020-12-21 11:33
 * @Version: 1.0
 */
@Service
@Slf4j
public class ConsumeDeductService {

    @Autowired
    private ConnectionFactory oneSyncConnectionFactory;
    @Autowired
    private Queue deductQueue; //扣费处理队列

    private Jackson2JsonMessageConverter jackson2JsonMessageConverter;

    public ConsumeDeductService(){
        jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    }

    /**
     * 接收并处理扣费消息 手动挡
     * 手动挡必须要自己设定参数,配置在application.properties中对于消费者的参数是无效的
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer handleDeductMessage() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(oneSyncConnectionFactory);
        container.setQueues(deductQueue);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);//默认就是自动
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            try {
                String correlationId = message.getMessageProperties().getCorrelationId();
                log.info("correlationId:{}", correlationId);
                Person deductMessage = (Person)jackson2JsonMessageConverter.fromMessage(message);//以Person存入mq就以Person强转

                log.info("Accept Deduct Message {}", deductMessage);
                if (!StringUtils.isEmpty(deductMessage)) {
                    throw new RuntimeException();
                }
            } catch (Exception e) {
                log.error("Receive Message Error", e);
                //告诉mq不回退队列,发往死信
                //throw new AmqpRejectAndDontRequeueException(e);
                throw new RuntimeException();//回退队列
            }
        });
        return container;
    }

    /**
     * 自动挡
     * @param message
     * @param channel
     * @throws IOException
     */
    //@RabbitListener(queues = "deductBusinessQueue")
    public void cumsumeDeductMessage(Message message, Channel channel) throws IOException {
        channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
    }

}